package com.activemq.protol;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.thread.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.boot.autoconfigure.jms.JmsProperties;

import javax.jms.*;

/**
 * 文档
 * https://activemq.apache.org/components/classic/documentation/configuring-version-5-transports.html
 * 参数文档：https://activemq.apache.org/components/classic/documentation/tcp-transport-reference
 * TCP、UDP、NIO、AMQP、SSL、MQTT
 * 默认消息协议是openwire
 * <transportConnector name="auto+nio" uri="auto+nio://0.0.0.0:61608?maximumConnections=1000&wireFormat.maxFrameSize=104857600&org.apache.activemq.transport.nio.SelectorManager.corePoolSize=20&org.apache.activemq.transport.nio.SelectorManager.maximumPoolSize=50"/>
 * <p>
 * NIO+多协议
 * auto
 * <p>
 * 不同协议，要编写的代码可能不相同,Mqtt需要使用MqttClient，需要引入新的jar包
 * auto直接包括tcp,amqp,mqtt协议
 */
public class MqProtolTestPlus {
    public static String PROTOCOL_URL = "failover://tcp://wangweiye:61608";

    public static String PROTOCOL_QUEUE = "AUTO_NIO_PROTOCOL_QUEUE";

    public static void main(String[] args) throws Exception {
        //&amp;org.apache.activemq.transport.nio.SelectorManager.corePoolSize=20&amp;org.apache.activemq.transport.nio.SelectorManager.maximumPoolSize=50
        //修改先修改activemq.xml，增加nio配置
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(PROTOCOL_URL);
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, JmsProperties.AcknowledgeMode.AUTO.getMode());
        Queue queue = session.createQueue(PROTOCOL_QUEUE);
        MessageProducer producer = session.createProducer(queue);
        for (int i = 0; i < 3; i++) {
            producer.send(session.createTextMessage("测试" + DateUtil.now()));
            ThreadUtil.safeSleep(1000);
        }
        producer.close();
        session.close();
        connection.close();
    }
}

@Slf4j
class MqConsumerPlus {
    public static void main(String[] args) throws Exception {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(MqProtolTestPlus.PROTOCOL_URL);
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, JmsProperties.AcknowledgeMode.AUTO.getMode());
        Queue queue = session.createQueue(MqProtolTestPlus.PROTOCOL_QUEUE);
        MessageConsumer consumer = session.createConsumer(queue);
        consumer.setMessageListener((message -> {
            try {
                if (message instanceof TextMessage) {
                    log.info("接收到消息：" + ((TextMessage) message).getText());
                }
            } catch (JMSException e) {
                log.error("出错", e);
            }
        }));
        System.in.read();
        consumer.close();
        session.close();
        connection.close();
    }
}